"""
@date: 2022/4/22 10:26
@author: kary
@desc: 算法引擎
"""
import os

from model_service.classify.torch_test import torch_classify_service
from model_service.segment.yolact_test import yolact_segment_service

os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
import json
from fastapi import FastAPI
import uvicorn
import time
import requests
import multiprocessing as mp
import mysql.connector as mysqlconnector
from model.model import *
from config import *
from SqlHelper import *
import datetime
from minio import Minio
import traceback
import urllib
from urllib.parse import urlparse
from algorithm.preprocess import *

from online_training.classify.step5_train import training
from online_training.classify.pytorch_classify.load_datatset import torch_load_datasets
from online_training.classify.pytorch_classify.train import torch_classify_train, torch_classify_train_pool
from model_service.classify.test import classify_service

# from online_training.detect.yolov5.step1_load_datatset import load_datasets
from online_training.detect.step1_load_datatset import load_datasets
from online_training.detect.yolov5.train import yolov5_train
from online_training.detect.EfficientDet.train import efficientdet_trian, efficientdet_trian_pool
from model_service.detect.step3_single_detect import detect_service
from model_service.detect.efficientdet_test import efficientdet_detect_service

from online_training.segment.maskrcnn_torch.load_dataset import load_segment_datasets
from online_training.segment.maskrcnn_torch.torch_train import maskrcnn_train_pool
from online_training.segment.yolact.step1_load_dataset import load_segment_yolact_datasets
from online_training.segment.yolact.train import yolact_train, yolact_train_pool
from model_service.segment.step4_test import segment_service

from algorithm.image_attributes import get_image_attributes
import psutil
import socket
import GPUtil
import sys

import logging

log = logging.getLogger('tensorflow')
log.setLevel(logging.DEBUG)

# create formatter and add it to the handlers
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# create file handler which logs even debug messages
fh = logging.FileHandler(os.path.join(os.path.dirname(os.path.realpath(__file__)),'log/aicv.log'))
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)
log.addHandler(fh)

app = FastAPI()

#使用封装的sqlhelper,通过使用数据库连接池解决模型训练太长时间，mysql服务器端口连接的bug
sqlhelper = MySQLhelper(dbhost=host, dbport='3306', dbuser=user, dbpassword=password, database=database)

def get_local_ip():
    local_ip = None
    try:
        hostname = socket.gethostname()  # 获取本机计算机名称
        local_ip = socket.gethostbyname(hostname)  # 获取本机ip
    except Exception as e:
        env_dist = os.environ
        local_ip = str(env_dist["HOSTIP"])
    return local_ip


# @app.post("/commitTask")
# async def commitTask(param: PreProcessTaskParam):
#     # print(param)
#     try:
#         # 写入数据库
#         # 设置数据库连接
#         db = mysqlconnector.connect(host=host, user=user, password=password, database=database, charset="utf8", buffered=True)
#         db.autocommit = True
#         cursor = db.cursor()
#
#         task_id = param.taskId
#         commit_time = param.commitTime
#         task_type = param.taskType
#         params = param.paramJson
#         update_url = param.updateUrl
#         other_update_url=param.otherUpdateUrl
#         run_ip = param.ip
#         node_type = param.nodeType
#
#         local_ip=get_local_ip(node_type)
#         print('run_ip',run_ip,'local_ip',local_ip)
#         if local_ip == run_ip:
#             print(local_ip + '新增一个任务')
#             cursor.execute("INSERT INTO biz_algorithm (taskId, commitTime, taskType, paramJson,taskUpdateUrl,otherUpdateUrl,taskStatus,ip,nodeType)" \
#                            "VALUES('{task_id}', '{commit_time}', '{task_type}', '{params}','{update_url}','{other_update_url}','{state}','{run_ip}','{node_type}')".format \
#                                (task_id=task_id, commit_time=commit_time, task_type=task_type, params=params, update_url=update_url, other_update_url=other_update_url,
#                                 state=TASK_STATUS_UNPROCESS,run_ip=run_ip,node_type=node_type))
#         else:
#             print(task_id,'非本节点任务！')
#         data = {"msg": "操作成功", "flag": 0}  # 成功
#     except Exception as e:
#         data = {"msg": "操作失败", "flag": 1, "detail": e}  # 操作失败
#     print(data)
#     return data


@app.post("/commitWeakLabelTask")
async def commitWeakLabelTask(param: WeakLabelTaskParam):
    try:
        url = param.url
        point_list = param.pointArr
        # 弱标注操作
        pass
        data = {"msg": "操作成功", "flag": 0, "result": 'test_success'}  # 成功
    except Exception as e:
        data = {"msg": "操作失败", "flag": 1, "detail": e}  # 操作失败
    return data


@app.post("/calculationAttribute")
async def calculationAttribute(param: ImageAttributesTaskParam):
    try:
        print('获取图像属性')
        images_list = param.imageAttributeROList
        MINIO_PATH_ALL = images_list[0]['imagePath']
        MINIO_HOST = MINIO_PATH_ALL.split('/')[2]
        MINIO_BUCKET_NAME = MINIO_PATH_ALL.split('/')[3]
        MINIO_BASE_URL = MINIO_PATH_ALL.split('/')[0] + '//' + MINIO_PATH_ALL.split('/')[2] + '/' + MINIO_PATH_ALL.split('/')[3] + '/'
        minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象
        result_list = []
        for img_dict in images_list:
            id = img_dict['id']
            image_minio_path = img_dict['imagePath']
            image_local_path = os.path.join(LOCAL_PATH, '图像属性', os.path.basename(urlparse(image_minio_path).path))  # 文件本地路径
            minio_path = image_minio_path.replace(MINIO_BASE_URL, "")
            minioClient.fget_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(minio_path), file_path=image_local_path)
            resolution, brightness, clearity = get_image_attributes(image_local_path)
            result = {'id': id, 'definition': clearity, 'brightness': brightness, 'resolvingPower': resolution}
            result_list.append(result)
        print(result_list)
        data = {"imageAttributeVOList": result_list}  # 成功
    except Exception as e:
        data = {"imageAttributeVOList": []}  # 操作失败
    return data


def execute_preprocess_task():
    local_ip = get_local_ip()  # 获取本机ip
    # 设置数据库连接
    # db = mysqlconnector.connect(host=host, user=user, password=password, database=database, charset="utf8", buffered=True)
    # db.autocommit = True
    # cursor = db.cursor()
    # 加载模型
    while (True):
        # 查询未执行预处理任务
        #cursor.execute("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus={state}  and taskType={task_type} and ip='{ipAdrr}' "
        #               "ORDER BY commitTime".format(state=TASK_STATUS_UNPROCESS, task_type=TASK_TYPE_PREPROCESS,ipAdrr=local_ip))
        #tasks = cursor.fetch_all()
        tasks = sqlhelper.fetch_all("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus=%s  and taskType=%s and ip=%s "
                       "ORDER BY commitTime", (TASK_STATUS_UNPROCESS, TASK_TYPE_PREPROCESS, local_ip))
        for current_task in tasks:
            # print(current_task)
            resultList = []
            dst_model_minio_path = ''

            try:
                task_id, commit_time, task_type, params, update_url, task_status, run_ip = current_task

                if local_ip != run_ip or local_ip is None:
                    continue
                print('task_id', task_id)
                start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                # cursor.execute("UPDATE biz_algorithm SET taskStartTime='{start_time}'  WHERE taskId='{task_id}'".format(start_time=start_time, task_id=task_id))
                sqlhelper.update("UPDATE biz_algorithm SET taskStartTime=%s  WHERE taskId=%s ", (start_time, task_id))

                params = json.loads(params)
                print('执行预处理任务')
                images_list = params['infoList']
                MINIO_PATH_ALL = images_list[0]['imagePath']
                MINIO_HOST = MINIO_PATH_ALL.split('/')[2]
                MINIO_BUCKET_NAME = MINIO_PATH_ALL.split('/')[3]
                MINIO_BASE_URL = MINIO_PATH_ALL.split('/')[0] + '//' + MINIO_PATH_ALL.split('/')[2] + '/' + MINIO_PATH_ALL.split('/')[3] + '/'
                # print('MINIO_HOST', MINIO_HOST)
                # print('MINIO_BASE_URL', MINIO_BASE_URL)
                minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象
                for img_dict in images_list:
                    image_minio_path = img_dict['imagePath']
                    suffix_name = os.path.basename(image_minio_path).split('.')[-1]
                    suffix_name = '.' + suffix_name
                    # print('image_minio_path', image_minio_path)
                    image_local_path = os.path.join(LOCAL_PATH, '预处理', os.path.basename(urlparse(image_minio_path).path))  # 文件本地路径
                    thumbnail_local_path = os.path.join(LOCAL_PATH, '预处理缩略图', os.path.basename(urlparse(image_minio_path).path))  # 缩略图本地路径

                    if not os.path.exists(os.path.dirname(thumbnail_local_path)):
                        os.makedirs(os.path.dirname(thumbnail_local_path))

                    minio_path = image_minio_path.replace(MINIO_BASE_URL, "")
                    # print('minio_path',minio_path)
                    minioClient.fget_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(minio_path), file_path=image_local_path)
                    # #生成原图缩略图
                    # thumbnail(image_local_path, thumbnail_local_path)  # 执行缩略图任务
                    # src_thumbnail_minio_path=image_minio_path.replace(img_minio_basename,'thumbnail_src/'+img_minio_basename)
                    # minioClient.fput_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(src_thumbnail_minio_path.replace(MINIO_BASE_URL, "")), file_path=thumbnail_local_path)  # 将原图缩略图的文件上传至minio

                    # 进行预处理操作
                    preprocessing(image_local_path)

                    with open(image_local_path, "rb") as f:
                        img_size = len(f.read())

                    ori_img_minio_basename = os.path.basename(urlparse(image_minio_path).path)

                    # print(img_minio_basename)

                    sstamp = str(time.time()).split('.')
                    new_pre = str(sstamp[0]) + str(sstamp[1][-3:])
                    img_minio_basename = ori_img_minio_basename.replace(ori_img_minio_basename.split('-')[0], new_pre)
                    # print(img_minio_basename)

                    dst_image_minio_path = image_minio_path.replace(ori_img_minio_basename, 'preprocess/' + img_minio_basename)
                    print('dst_image_minio_path', dst_image_minio_path)
                    minioClient.fput_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(dst_image_minio_path.replace(MINIO_BASE_URL, "")), file_path=image_local_path,
                                            content_type="application/x-www-form-urlencoded")  # 将处理后的文件上传至minio

                    # 生成预处理后的图的缩略图
                    thumbnail(image_local_path, thumbnail_local_path)  # 执行缩略图任务
                    dst_thumbnail_minio_path = image_minio_path.replace(ori_img_minio_basename, 'thumbnail/' + img_minio_basename)
                    # print('dst_thumbnail_minio_path', dst_thumbnail_minio_path)
                    minioClient.fput_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(dst_thumbnail_minio_path.replace(MINIO_BASE_URL, "")),
                                            file_path=thumbnail_local_path)  # 将处理后的缩略图文件上传至minio

                    result = {'id': img_dict['id'], 'imagePath': dst_image_minio_path, 'name': img_dict['name'], 'thumbnailPath': dst_thumbnail_minio_path, 'imageSize': img_size}
                    resultList.append(result)

                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                data = {'id': task_id, 'startTime': start_time, 'modelPath': dst_model_minio_path, 'infoList': resultList, 'endTime': end_time, 'result': None, 'status': STATUS_SUCCESS}
            except Exception as e:
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                error_info = traceback.format_exc()
                data = {'id': task_id, 'startTime': start_time, 'modelPath': None, 'infoList': resultList, 'endTime': end_time, 'result': str(error_info), 'status': STATUS_FAIL}
                # log_path = os.path.join('log', end_time.split(' ')[0] + '_preprocess_log.txt')
                log_path = os.path.join(os.path.dirname(os.path.realpath(__file__)),os.path.join('log', end_time.split(' ')[0] + '_preprocess_log.txt'))
                with open(log_path, "a+") as f:
                    f.writelines("{} {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), error_info))
            # t=json.dumps(data)
            response = requests.post(update_url, headers=HEADERS, timeout=20, data=json.dumps(data))
            #cursor.execute("UPDATE biz_algorithm SET taskStatus='{state}',taskEndTime='{end_time}'  WHERE taskId='{task_id}'".format(state=TASK_STATUS_PROCESSED, end_time=end_time, task_id=task_id))
            sqlhelper.update("UPDATE biz_algorithm SET taskStatus=%s,taskEndTime=%s  WHERE taskId=%s ", (TASK_STATUS_PROCESSED, end_time, task_id))
            print('done!')
        else:
            time.sleep(3)


def execute_labeling_task():
    local_ip = get_local_ip()  # 获取本机ip
    # 设置数据库连接
    # db = mysqlconnector.connect(host=host, user=user, password=password, database=database, charset="utf8", buffered=True)
    # db.autocommit = True
    # cursor = db.cursor()
    # 加载模型
    while (True):
        # 查询未执行的标注任务
        #cursor.execute("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus={state} and taskType={task_type} and ip='{ipAdrr}' "
        #               "ORDER BY commitTime".format(state=TASK_TYPE_LABELING, task_type=TASK_TYPE_LABELING, ipAdrr=local_ip))
        #tasks = cursor.fetchall()
        tasks = sqlhelper.fetch_all("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus=%s and taskType=%s and ip=%s "
                       "ORDER BY commitTime", (TASK_TYPE_LABELING, TASK_TYPE_LABELING, local_ip))
        for current_task in tasks:
            # print(current_task)
            resultList = []
            dst_model_minio_path = ''
            try:
                task_id, commit_time, task_type, params, update_url, task_status, run_ip = current_task

                if local_ip != run_ip or local_ip is None:
                    continue
                print('task_id', task_id)
                start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                # cursor.execute("UPDATE biz_algorithm SET taskStartTime='{start_time}'  WHERE taskId='{task_id}'".format(start_time=start_time, task_id=task_id))
                sqlhelper.update("UPDATE biz_algorithm SET taskStartTime=%s  WHERE taskId=%s ", (start_time, task_id))

                params = json.loads(params)
                pass  # 标注
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                data = {'id': task_id, 'startTime': start_time, 'modelPath': dst_model_minio_path, 'infoList': resultList, 'endTime': end_time, 'result': None, 'status': STATUS_SUCCESS}
            except Exception as e:
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                error_info = traceback.format_exc()
                data = {'id': task_id, 'startTime': start_time, 'modelPath': None, 'infoList': resultList, 'endTime': end_time, 'result': str(error_info), 'status': STATUS_FAIL}
                # log_path = os.path.join('log', end_time.split(' ')[0] + '_labeling_log.txt')
                log_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), os.path.join('log', end_time.split(' ')[0] + '_labeling_log.txt'))
                with open(log_path, "a+") as f:
                    f.writelines("{} {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), error_info))
            # t=json.dumps(data)
            response = requests.post(update_url, headers=HEADERS, timeout=10, data=json.dumps(data))
            # cursor.execute("UPDATE biz_algorithm SET taskStatus='{state}',taskEndTime='{end_time}'  WHERE taskId='{task_id}'".format(state=TASK_STATUS_PROCESSED, end_time=end_time, task_id=task_id))
            sqlhelper.update("UPDATE biz_algorithm SET taskStatus=%s,taskEndTime=%s  WHERE taskId=%s ",(TASK_STATUS_PROCESSED, end_time, task_id))
            print('done!')
        else:
            time.sleep(3)


def execute_training_task():
    local_ip = get_local_ip()  # 获取本机ip
    # 设置数据库连接
    # db = mysqlconnector.connect(host=host, user=user, password=password, database=database, charset="utf8", buffered=True)
    # db.autocommit = True
    # cursor = db.cursor()
    # 加载模型
    while (True):
        # 查询未执行预处理任务
        # cursor.execute("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,taskStatus,ip,resumeTrain FROM biz_algorithm  where taskStatus={state} and taskType={task_type} and ip='{ipAdrr}' "
        #               "ORDER BY commitTime".format(state=TASK_STATUS_UNPROCESS, task_type=TASK_TYPE_TRAINING, ipAdrr=local_ip))
        # tasks = cursor.fetchall()
        tasks = sqlhelper.fetch_all("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,taskStatus,ip,resumeTrain FROM biz_algorithm  where taskStatus=%s and taskType=%s and ip=%s "
                       "ORDER BY commitTime", (TASK_STATUS_UNPROCESS, TASK_TYPE_TRAINING, local_ip))
        for current_task in tasks:
            resultList = []
            dst_model_minio_path = ''
            userId = ''
            start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            try:
                task_id, commit_time, task_type, params, update_url, task_status, run_ip, resumeTrain = current_task

                if local_ip != run_ip or local_ip is None:
                    continue
                print('task_id', task_id)
                # cursor.execute("UPDATE biz_algorithm SET taskStartTime='{start_time}'  WHERE taskId='{task_id}'".format(start_time=start_time, task_id=task_id))
                sqlhelper.update("UPDATE biz_algorithm SET taskStartTime=%s,resumeTrain = 1  WHERE taskId=%s", (start_time, task_id))

                params = json.loads(params)
                userId = params['userId']
                model_type = params['type']
                if model_type == MODEL_TYPE_CLASSIFY:
                    # 获取数据集
                    print('执行分类在线训练任务')
                    trainList = params['trainDataset']
                    testList = params['testDataset']
                    trainParams = params['trainParams']
                    labelList = trainParams['labelList']
                    image_minio_path = trainList[0]['imagePath']

                    MINIO_HOST = image_minio_path.split('/')[2]
                    MINIO_BUCKET_NAME = image_minio_path.split('/')[3]
                    MINIO_BASE_URL = image_minio_path.split('/')[0] + '//' + image_minio_path.split('/')[2] + '/' + image_minio_path.split('/')[3] + '/'
                    minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象
                    if trainParams['netName'] == 'InceptionV3' or trainParams['netName'] == 'Xception' or trainParams['netName'] == 'ResNet101':
                        result_model_path, topk_acc, p, r, f1score, validity = training(trainList, testList, minioClient, MINIO_BASE_URL, MINIO_BUCKET_NAME, trainParams)
                    elif trainParams['netName'] == 'DenseNet121' or trainParams['netName'] == 'PyramidNet200' or trainParams['netName'] == 'Resnext50':
                        dataset_path = torch_load_datasets(trainList, testList, labelList, minioClient, MINIO_BASE_URL, MINIO_BUCKET_NAME, trainParams)
                        # result_model_path, topk_acc, p, r, f1score, validity = torch_classify_train(dataset_path, trainParams)
                        result_model_path, topk_acc, p, r, f1score, validity, ex = torch_classify_train_pool(dataset_path, trainParams)

                    model_basename = os.path.basename(urlparse(result_model_path).path)
                    ori_img_minio_basename = os.path.basename(urlparse(image_minio_path).path)
                    if result_model_path is not None:
                        print('topk_acc, p, r, f1score, validity', topk_acc, p, r, f1score, validity)
                        dst_model_minio_path = image_minio_path.replace(ori_img_minio_basename, 'classify_models/' + model_basename)
                        minioClient.fput_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(dst_model_minio_path.replace(MINIO_BASE_URL, "")), file_path=result_model_path,
                                                content_type="application/x-www-form-urlencoded")  # 将训练好的模型上传至minio
                        status = TRAIN_SUCCESS

                        shutil.rmtree(dataset_path)
                        # shutil.rmtree(os.path.dirname(result_model_path))

                    else:
                        dst_model_minio_path = None
                        status = TRAIN_FAIL
                        if ex is not None:
                            ex = str(ex)
                    print('dst_model_minio_path', dst_model_minio_path)
                    resultList = {'topkAcc': topk_acc, 'precision': p, 'recall': r, 'f1Score': f1score}

                    end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                    data = {'id': task_id, 'userId': userId, 'startTime': start_time, 'modelPath': dst_model_minio_path, 'validity': validity, 'infoList': resultList, 'infoList1': None,
                            'infoList2': None,
                            'endTime': end_time, 'result': ex,
                            'status': status}

                if model_type == MODEL_TYPE_DETECT:  # 检测
                    print('执行检测在线训练任务')
                    trainList = params['trainDataset']
                    testList = params['testDataset']
                    trainParams = params['trainParams']

                    image_minio_path = trainList[0]['imagePath']

                    MINIO_HOST = image_minio_path.split('/')[2]
                    MINIO_BUCKET_NAME = image_minio_path.split('/')[3]
                    MINIO_BASE_URL = image_minio_path.split('/')[0] + '//' + image_minio_path.split('/')[2] + '/' + image_minio_path.split('/')[3] + '/'
                    minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象
                    yolov5_format_dir, coco_format_dir, voc_format_dir, label_names, test_data_dir = load_datasets(trainList, testList, minioClient, MINIO_BASE_URL, MINIO_BUCKET_NAME, trainParams)
                    if trainParams['netName'] == 'YOLOv5':
                        trainParams['resume'] = resumeTrain
                        # 训练
                        results, save_model_path = yolov5_train(yolov5_format_dir, label_names, trainParams)  # (mp, mr, map50, map, *(loss.cpu() / len(dataloader)).tolist()), maps, t
                        # results, save_model_path = yolov5_train_pool(yolov5_format_dir, label_names, trainParams)
                        # results, save_model_path=(0.0, 0.0, 0.0, 0.0, 0.06270400434732437, 0.028786275535821915, 0.023556426167488098),r'D:\python_project\aicv_python/online_training\detect\yolov5\runs\train\exp/weights/best.pt'

                        print(results, save_model_path)
                        n1 = os.path.basename(os.path.dirname(os.path.dirname(save_model_path)))
                        n2 = os.path.basename(save_model_path)
                        suffix_model_name = n1 + '_' + n2
                        model_basename = os.path.basename(os.path.dirname(yolov5_format_dir)) + '_' + suffix_model_name

                        ori_img_minio_basename = os.path.basename(urlparse(image_minio_path).path)
                        dst_model_minio_path = image_minio_path.replace(ori_img_minio_basename, 'detect_models/' + model_basename)
                        print('dst_model_minio_path', dst_model_minio_path)
                        minioClient.fput_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(dst_model_minio_path.replace(MINIO_BASE_URL, "")), file_path=save_model_path,
                                                content_type="application/x-www-form-urlencoded")  # 将训练好的模型上传至minio

                        resultList = {'map': float(round(results[3], 3)), 'precision': float(round(results[0], 3)), 'recall': float(round(results[1], 3)), 'classificationLoss': None}
                        validity = float(round(results[3], 3))

                        # shutil.rmtree(os.path.dirname(os.path.dirname(os.path.dirname(save_model_path))))
                        shutil.rmtree(os.path.dirname(yolov5_format_dir))
                        end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

                        data = {'id': task_id, 'userId': userId, 'startTime': start_time, 'modelPath': dst_model_minio_path, 'validity': validity, 'infoList': None, 'infoList1': resultList,
                                'infoList2': None, 'endTime': end_time,
                                'result': None,
                                'status': TRAIN_SUCCESS}
                    elif trainParams['netName'] == 'EfficientDet':
                        trainParams['load_weights'] = str(resumeTrain)
                        # results, save_model_path = efficientdet_trian(coco_format_dir, label_names, trainParams)
                        results, save_model_path, ex = efficientdet_trian_pool(coco_format_dir, label_names, trainParams)
                        # print(results, save_model_path)
                        if save_model_path is not None:
                            n2 = os.path.basename(save_model_path)
                            suffix_model_name = n2
                            model_basename = os.path.basename(os.path.dirname(coco_format_dir)) + '_' + suffix_model_name

                            ori_img_minio_basename = os.path.basename(urlparse(image_minio_path).path)
                            dst_model_minio_path = image_minio_path.replace(ori_img_minio_basename, 'detect_models/' + model_basename)
                            print('dst_model_minio_path', dst_model_minio_path)
                            minioClient.fput_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(dst_model_minio_path.replace(MINIO_BASE_URL, "")), file_path=save_model_path,
                                                    content_type="application/x-www-form-urlencoded")  # 将训练好的模型上传至minio

                            resultList = {'map': float(round(results[1], 3)), 'precision': float(round(results[1], 3)), 'recall': float(round(results[8], 3)), 'classificationLoss': None}
                            validity = float(round(results[1], 3))

                            # shutil.rmtree(os.path.dirname(save_model_path))
                            shutil.rmtree(os.path.dirname(coco_format_dir))
                            end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

                            data = {'id': task_id, 'userId': userId, 'startTime': start_time, 'modelPath': dst_model_minio_path, 'validity': validity, 'infoList': None, 'infoList1': resultList,
                                    'infoList2': None, 'endTime': end_time,
                                    'result': None,
                                    'status': TRAIN_SUCCESS}
                        else:
                            dst_model_minio_path = None
                            status = TRAIN_FAIL
                            if ex is not None:
                                ex = str(ex)
                            end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                            validity = None
                            resultList = {}
                            data = {'id': task_id, 'userId': userId, 'startTime': start_time, 'modelPath': dst_model_minio_path, 'validity': validity, 'infoList': resultList, 'infoList1': None,
                                    'infoList2': None,
                                    'endTime': end_time, 'result': ex,
                                    'status': status}

                if model_type == MODEL_TYPE_SEGMENT:  # 分割
                    print('执行分割在线训练任务')
                    trainList = params['trainDataset']
                    testList = params['testDataset']
                    trainParams = params['trainParams']
                    label_names = trainParams['labelList']
                    image_minio_path = trainList[0]['imagePath']

                    MINIO_HOST = image_minio_path.split('/')[2]
                    MINIO_BUCKET_NAME = image_minio_path.split('/')[3]
                    MINIO_BASE_URL = image_minio_path.split('/')[0] + '//' + image_minio_path.split('/')[2] + '/' + image_minio_path.split('/')[3] + '/'
                    minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象

                    mIOU = 0
                    validity = None
                    save_model_path = None
                    train_data_dir,train_dataset_path=None,None
                    if trainParams['netName'] == 'Mask R-CNN':
                        train_data_dir = load_segment_datasets(trainList, testList, minioClient, MINIO_BASE_URL, MINIO_BUCKET_NAME, trainParams, label_names)
                        mIOU, save_model_path, ex = maskrcnn_train_pool(train_data_dir, trainParams)
                    elif trainParams['netName'] == 'Yolact':
                        train_dataset_path, class_names = load_segment_yolact_datasets(trainList, testList, minioClient, MINIO_BASE_URL, MINIO_BUCKET_NAME, trainParams, label_names)
                        # mIOU, validity, save_model_path = yolact_train(train_dataset_path, class_names, trainParams)
                        mIOU, validity, save_model_path, ex = yolact_train_pool(train_dataset_path, class_names, trainParams)
                    if save_model_path is not None:
                        print(mIOU, save_model_path)

                        # mIOU=0.8455934897318702
                        # save_model_path=r'D:\aicv_data\segment_online_training\dataset\1664429201932data\save_models\1664429201932maskrcnn.pth'
                        # dst_model_minio_path = r'http://124.70.140.175:9000/aicv/52f0338a-d53a-4e90-9516-f68c4464bf73(鱼生长态势分析-小架子-分割数据集)/2022-09-03/segment_models/1664429201932maskrcnn.pth'
                        # mIOU = 0.8455934897318702

                        model_basename = os.path.basename(save_model_path)
                        ori_img_minio_basename = os.path.basename(urlparse(image_minio_path).path)
                        dst_model_minio_path = image_minio_path.replace(ori_img_minio_basename, 'segment_models/' + model_basename)

                        print('dst_model_minio_path', dst_model_minio_path)
                        minioClient.fput_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(dst_model_minio_path.replace(MINIO_BASE_URL, "")), file_path=save_model_path,
                                                content_type="application/x-www-form-urlencoded")  # 将训练好的模型上传至minio
                        resultList = {'miou': float(round(mIOU, 3))}
                        validity = float(round(mIOU, 3)) if validity is None else validity
                        if train_data_dir is not None:
                            shutil.rmtree(os.path.dirname(train_data_dir))  #删除数据集
                        if train_dataset_path is not None:
                            shutil.rmtree(os.path.dirname(train_dataset_path))
                        if os.path.exists(save_model_path):
                            # shutil.rmtree(os.path.dirname(save_model_path))    #删除模型
                            pass

                        end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                        data = {'id': task_id, 'userId': userId, 'startTime': start_time, 'modelPath': dst_model_minio_path, 'validity': validity, 'infoList': None, 'infoList1': None,
                                'infoList2': resultList, 'endTime': end_time, 'result': None,
                                'status': TRAIN_SUCCESS}
                    else:
                        end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                        data = {'id': task_id, 'userId': userId, 'startTime': start_time, 'modelPath': None, 'validity': None, 'infoList': None, 'infoList1': None, 'infoList2': None,
                                'endTime': end_time, 'result': str(ex),
                                'status': TRAIN_FAIL}

            except Exception as e:
                print(e)
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                error_info = traceback.format_exc()
                data = {'id': task_id, 'userId': userId, 'startTime': start_time, 'modelPath': None, 'validity': None, 'infoList': None, 'infoList1': None, 'infoList2': None,
                        'endTime': end_time, 'result': str(error_info),
                        'status': TRAIN_FAIL}
                # log_path = os.path.join('log', end_time.split(' ')[0] + '_training_log.txt')
                log_path = os.path.join(os.path.dirname(os.path.realpath(__file__)),os.path.join('log', end_time.split(' ')[0] + '_training_log.txt'))
                with open(log_path, "a+") as f:
                    f.writelines("{} {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), error_info))
            print(data)
            # print(json.dumps(data))
            response = requests.post(update_url, headers=HEADERS, timeout=10, data=json.dumps(data))
            print(response)
            # cursor.execute("UPDATE biz_algorithm SET taskStatus='{state}',taskEndTime='{end_time}'  WHERE taskId='{task_id}'".format(state=TASK_STATUS_PROCESSED, end_time=end_time, task_id=task_id))
            sqlhelper.update("UPDATE biz_algorithm SET taskStatus=%s,taskEndTime=%s  WHERE taskId=%s ",(TASK_STATUS_PROCESSED, end_time, task_id))
            print('done!')
        else:
            time.sleep(3)


def classify_service_task():
    local_ip = get_local_ip()  # 获取本机ip
    # 设置数据库连接
    # db = mysqlconnector.connect(host=host, user=user, password=password, database=database, charset="utf8", buffered=True)
    # db.autocommit = True
    # cursor = db.cursor()
    # 加载模型
    while (True):
        # 查询未执行的任务
        # cursor.execute("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,otherUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus={state} and taskType={task_type} and ip='{ipAdrr}' "
        #               "ORDER BY commitTime".format(state=TASK_STATUS_UNPROCESS, task_type=TASK_TYPE_CLASSIFICATION,ipAdrr=local_ip))
        # tasks = cursor.fetchall()
        tasks = sqlhelper.fetch_all("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,otherUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus=%s and taskType=%s and ip=%s "
                       "ORDER BY commitTime", (TASK_STATUS_UNPROCESS, TASK_TYPE_CLASSIFICATION,local_ip))
        for current_task in tasks:
            # print(current_task)
            resultList = []
            dst_model_minio_path = ''
            log_id = ''
            try:
                task_id, commit_time, task_type, params, update_url, other_update_url, task_status, run_ip = current_task

                if local_ip != run_ip or local_ip is None:
                    continue
                print('task_id', task_id)
                start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                # cursor.execute("UPDATE biz_algorithm SET taskStartTime='{start_time}'  WHERE taskId='{task_id}'".format(start_time=start_time, task_id=task_id))
                sqlhelper.update("UPDATE biz_algorithm SET taskStartTime=%s  WHERE taskId=%s ", (start_time, task_id))

                params = json.loads(params)
                print('执行分类模型服务')

                log_id = params['logId']
                # log_id=''
                # 下载模型
                model_minio_path = params['modelPath']
                MINIO_HOST = model_minio_path.split('/')[2]
                MINIO_BUCKET_NAME = model_minio_path.split('/')[3]
                MINIO_BASE_URL = model_minio_path.split('/')[0] + '//' + model_minio_path.split('/')[2] + '/' + model_minio_path.split('/')[3] + '/'
                minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象
                model_local_path = os.path.join(LOCAL_PATH, '分类服务/模型', os.path.basename(urlparse(model_minio_path).path))  # 模型本地路径
                minio_path = model_minio_path.replace(MINIO_BASE_URL, "")
                minioClient.fget_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(minio_path), file_path=model_local_path)

                image_list = params['infoList']
                MINIO_PATH_ALL = image_list[0]['path']
                MINIO_HOST = MINIO_PATH_ALL.split('/')[2]
                MINIO_BUCKET_NAME = MINIO_PATH_ALL.split('/')[3]
                MINIO_BASE_URL = MINIO_PATH_ALL.split('/')[0] + '//' + MINIO_PATH_ALL.split('/')[2] + '/' + MINIO_PATH_ALL.split('/')[3] + '/'
                minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象

                image_path_list = []
                image_id_list = []
                for img_dict in image_list:
                    id = img_dict['id']
                    image_id_list.append(id)

                    image_minio_path = img_dict['path']
                    image_local_path = os.path.join(LOCAL_PATH, '分类服务/图像', os.path.basename(urlparse(image_minio_path).path))  # 文件本地路径
                    minio_path = image_minio_path.replace(MINIO_BASE_URL, "")
                    minioClient.fget_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(minio_path), file_path=image_local_path)
                    image_path_list.append(image_local_path)

                train_params = params['trainParams']  # 模型训练时的参数
                # resultList = classify_service(model_local_path, image_path_list, image_list, image_id_list, train_params)  # {名字：标签}形式的键值对列表
                resultList = torch_classify_service(model_local_path, image_path_list, image_list, image_id_list, train_params)
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                data = {'id': log_id, 'dataList': resultList, 'errMessage': None, 'status': CLASSIFY_SERVICE_SUCCESS, 'startTime': start_time, 'endTime': end_time}
            except Exception as e:
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                error_info = traceback.format_exc()
                data = {'id': log_id, 'dataList': resultList, 'errMessage': str(error_info), 'status': CLASSIFY_SERVICE_FAIL, 'startTime': start_time, 'endTime': end_time}

                log_path = os.path.join(os.path.dirname(os.path.realpath(__file__)),os.path.join('log', end_time.split(' ')[0] + '_classify_service_log.txt'))
                with open(log_path, "a+") as f:
                    f.writelines("{} {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), error_info))
            # t=json.dumps(data)
            print(json.dumps(data))

            response = requests.post(update_url, headers=HEADERS, timeout=10, data=json.dumps(data))
            print('回调地址', update_url, response)
            if other_update_url is not None:
                response = requests.post(other_update_url, headers=HEADERS, timeout=10, data=json.dumps(data))
                print('其他回调地址', other_update_url, response)
            # cursor.execute("UPDATE biz_algorithm SET taskStatus='{state}',taskEndTime='{end_time}'  WHERE taskId='{task_id}'".format(state=TASK_STATUS_PROCESSED, end_time=end_time, task_id=task_id))
            sqlhelper.update("UPDATE biz_algorithm SET taskStatus=%s,taskEndTime=%s  WHERE taskId=%s",(TASK_STATUS_PROCESSED, end_time, task_id))

            print('done!')
        else:
            time.sleep(3)


def detect_service_task():
    local_ip = get_local_ip()  # 获取本机ip
    # 设置数据库连接
    # db = mysqlconnector.connect(host=host, user=user, password=password, database=database, charset="utf8", buffered=True)
    # db.autocommit = True
    # cursor = db.cursor()
    # 加载模型
    while (True):
        # 查询未执行的检测任务
        # cursor.execute("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,otherUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus={state} and taskType={task_type} and ip='{ipAdrr}' "
        #               "ORDER BY commitTime".format(state=TASK_STATUS_UNPROCESS, task_type=TASK_TYPE_DETECTION, ipAdrr=local_ip))
        # tasks = cursor.fetchall()
        tasks = sqlhelper.fetch_all("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,otherUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus=%s and taskType=%s and ip=%s "
                       "ORDER BY commitTime", (TASK_STATUS_UNPROCESS, TASK_TYPE_DETECTION, local_ip))
        for current_task in tasks:
            resultList = []
            dst_model_minio_path = ''
            log_id = ''
            try:
                task_id, commit_time, task_type, params, update_url, other_update_url, task_status, run_ip = current_task

                if local_ip != run_ip or local_ip is None:
                    continue
                print('task_id', task_id)
                start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                # cursor.execute("UPDATE biz_algorithm SET taskStartTime='{start_time}'  WHERE taskId='{task_id}'".format(start_time=start_time, task_id=task_id))
                sqlhelper.update("UPDATE biz_algorithm SET taskStartTime=%s  WHERE taskId=%s ", (start_time, task_id))

                params = json.loads(params)
                print('执行检测模型服务')

                log_id = params['logId']

                # 下载模型
                model_minio_path = params['modelPath']
                MINIO_HOST = model_minio_path.split('/')[2]
                MINIO_BUCKET_NAME = model_minio_path.split('/')[3]
                MINIO_BASE_URL = model_minio_path.split('/')[0] + '//' + model_minio_path.split('/')[2] + '/' + model_minio_path.split('/')[3] + '/'
                minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象

                model_local_path = os.path.join(LOCAL_PATH, '检测服务/模型', os.path.basename(urlparse(model_minio_path).path))  # 模型本地路径
                minio_path = model_minio_path.replace(MINIO_BASE_URL, "")
                minioClient.fget_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(minio_path), file_path=model_local_path)

                image_list = params['infoList']
                MINIO_PATH_ALL = image_list[0]['path']
                MINIO_HOST = MINIO_PATH_ALL.split('/')[2]
                MINIO_BUCKET_NAME = MINIO_PATH_ALL.split('/')[3]
                MINIO_BASE_URL = MINIO_PATH_ALL.split('/')[0] + '//' + MINIO_PATH_ALL.split('/')[2] + '/' + MINIO_PATH_ALL.split('/')[3] + '/'
                minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象

                image_path_list = []
                image_id_list = []
                for img_dict in image_list:
                    id = img_dict['id']
                    image_id_list.append(id)

                    image_minio_path = img_dict['path']
                    image_local_path = os.path.join(LOCAL_PATH, '检测服务/图像', os.path.basename(urlparse(image_minio_path).path))  # 文件本地路径
                    minio_path = image_minio_path.replace(MINIO_BASE_URL, "")
                    minioClient.fget_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(minio_path), file_path=image_local_path)
                    image_path_list.append(image_local_path)

                train_params = params['trainParams']  # 模型训练时的参数
                if train_params['netName'] == 'YOLOv5':
                    resultList = detect_service(model_local_path, image_path_list, image_list, image_id_list, train_params)  # {名字：标签}形式的键值对列表
                elif train_params['netName'] == 'EfficientDet':
                    resultList = efficientdet_detect_service(model_local_path, image_path_list, image_list, image_id_list, train_params)  # {名字：标签}形式的键值对列表

                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                data = {'id': log_id, 'dataList': resultList, 'errMessage': None, 'status': CLASSIFY_SERVICE_SUCCESS, 'startTime': start_time, 'endTime': end_time}
            except Exception as e:
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                error_info = traceback.format_exc()
                data = {'id': log_id, 'dataList': resultList, 'errMessage': str(error_info), 'status': CLASSIFY_SERVICE_FAIL, 'startTime': start_time, 'endTime': end_time}

                # log_path = os.path.join('log', end_time.split(' ')[0] + '_classify_service_log.txt')
                log_path = os.path.join(os.path.dirname(os.path.realpath(__file__)),os.path.join('log', end_time.split(' ')[0] + '_classify_service_log.txt'))
                with open(log_path, "a+") as f:
                    f.writelines("{} {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), error_info))
            # t=json.dumps(data)
            print(json.dumps(data))
            response = requests.post(update_url, headers=HEADERS, timeout=10, data=json.dumps(data))
            print('回调地址', update_url, response)
            if other_update_url is not None:
                response = requests.post(other_update_url, headers=HEADERS, timeout=10, data=json.dumps(data))
                print('其他回调地址', other_update_url, response)
            # cursor.execute("UPDATE biz_algorithm SET taskStatus='{state}',taskEndTime='{end_time}'  WHERE taskId='{task_id}'".format(state=TASK_STATUS_PROCESSED, end_time=end_time, task_id=task_id))
            sqlhelper.update("UPDATE biz_algorithm SET taskStatus=%s,taskEndTime=%s  WHERE taskId=%s ",(TASK_STATUS_PROCESSED, end_time, task_id))

            print('done!')
        else:
            time.sleep(3)


def segment_service_task():
    local_ip = get_local_ip()  # 获取本机ip
    # 设置数据库连接
    # db = mysqlconnector.connect(host=host, user=user, password=password, database=database, charset="utf8", buffered=True)
    # db.autocommit = True
    # cursor = db.cursor()
    # 加载模型
    while (True):
        # 查询未执行的分割任务
        # cursor.execute("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,otherUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus={state} and taskType={task_type} and ip='{ipAdrr}'  "
        #               "ORDER BY commitTime".format(state=TASK_STATUS_UNPROCESS, task_type=TASK_TYPE_SEGMENTATION, ipAdrr=local_ip))
        # tasks = cursor.fetchall()
        tasks = sqlhelper.fetch_all("SELECT taskId, commitTime, taskType, paramJson,taskUpdateUrl,otherUpdateUrl,taskStatus,ip FROM biz_algorithm  where taskStatus=%s and taskType=%s and ip=%s  "
                       "ORDER BY commitTime" ,(TASK_STATUS_UNPROCESS, TASK_TYPE_SEGMENTATION, local_ip))
        for current_task in tasks:
            resultList = []
            dst_model_minio_path = ''
            log_id = ''
            try:
                task_id, commit_time, task_type, params, update_url, other_update_url, task_status, run_ip = current_task

                if local_ip != run_ip or local_ip is None:
                    continue
                print('task_id', task_id)
                start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                # cursor.execute("UPDATE biz_algorithm SET taskStartTime='{start_time}'  WHERE taskId='{task_id}'".format(start_time=start_time, task_id=task_id))
                sqlhelper.update("UPDATE biz_algorithm SET taskStartTime=%s  WHERE taskId=%s  ", (start_time, task_id))

                params = json.loads(params)
                print('执行分割模型服务')

                log_id = params['logId']

                # 下载模型
                model_minio_path = params['modelPath']
                MINIO_HOST = model_minio_path.split('/')[2]
                MINIO_BUCKET_NAME = model_minio_path.split('/')[3]
                MINIO_BASE_URL = model_minio_path.split('/')[0] + '//' + model_minio_path.split('/')[2] + '/' + model_minio_path.split('/')[3] + '/'
                minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象

                model_local_path = os.path.join(LOCAL_PATH, '分割服务/模型', os.path.basename(urlparse(model_minio_path).path))  # 模型本地路径
                minio_path = model_minio_path.replace(MINIO_BASE_URL, "")
                minioClient.fget_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(minio_path), file_path=model_local_path)

                image_list = params['infoList']
                MINIO_PATH_ALL = image_list[0]['path']
                MINIO_HOST = MINIO_PATH_ALL.split('/')[2]
                MINIO_BUCKET_NAME = MINIO_PATH_ALL.split('/')[3]
                MINIO_BASE_URL = MINIO_PATH_ALL.split('/')[0] + '//' + MINIO_PATH_ALL.split('/')[2] + '/' + MINIO_PATH_ALL.split('/')[3] + '/'
                minioClient = Minio(endpoint=MINIO_HOST, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=False)  # 新建minio客户端对象

                image_path_list = []
                image_id_list = []
                for img_dict in image_list:
                    id = img_dict['id']
                    image_id_list.append(id)

                    image_minio_path = img_dict['path']
                    image_local_path = os.path.join(LOCAL_PATH, '分割服务/图像', os.path.basename(urlparse(image_minio_path).path))  # 文件本地路径
                    minio_path = image_minio_path.replace(MINIO_BASE_URL, "")
                    minioClient.fget_object(bucket_name=MINIO_BUCKET_NAME, object_name=urllib.parse.unquote(minio_path), file_path=image_local_path)
                    image_path_list.append(image_local_path)

                trainParams = params['trainParams']  # 模型训练时的参数
                if trainParams['netName'] == 'Mask R-CNN':
                    resultList = segment_service(model_local_path, image_path_list, image_list, image_id_list, trainParams)  # {名字：标签}形式的键值对列表
                elif trainParams['netName'] == 'Yolact':
                    resultList = yolact_segment_service(model_local_path, image_path_list, image_list, image_id_list, trainParams)
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                data = {'id': log_id, 'dataList': resultList, 'errMessage': None, 'status': CLASSIFY_SERVICE_SUCCESS, 'startTime': start_time, 'endTime': end_time}
            except Exception as e:
                end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                error_info = traceback.format_exc()
                data = {'id': log_id, 'dataList': resultList, 'errMessage': str(error_info), 'status': CLASSIFY_SERVICE_FAIL, 'startTime': start_time, 'endTime': end_time}

                # log_path = os.path.join('log', end_time.split(' ')[0] + '_segment_service_log.txt')
                log_path = os.path.join(os.path.dirname(os.path.realpath(__file__)),os.path.join('log', end_time.split(' ')[0] + '_segment_service_log.txt'))
                with open(log_path, "a+") as f:
                    f.writelines("{} {}\n".format(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), error_info))
            # t=json.dumps(data)
            print(json.dumps(data))
            response = requests.post(update_url, headers=HEADERS, timeout=10, data=json.dumps(data))
            print('回调地址', update_url, response)
            if other_update_url is not None:
                response = requests.post(other_update_url, headers=HEADERS, timeout=10, data=json.dumps(data))
                print('其他回调地址', other_update_url, response)
            # cursor.execute("UPDATE biz_algorithm SET taskStatus='{state}',taskEndTime='{end_time}'  WHERE taskId='{task_id}'".format(state=TASK_STATUS_PROCESSED, end_time=end_time, task_id=task_id))
            sqlhelper.update("UPDATE biz_algorithm SET taskStatus=%s,taskEndTime=%s  WHERE taskId=%s ",(TASK_STATUS_PROCESSED, end_time, task_id))
            print('done!')
        else:
            time.sleep(3)


def performance_monitoring_task():
    while True:  # 轮询执行任务
        time_start = time.time()
        try:
            cpu_per = round((psutil.cpu_percent(interval=1)) / 100, 2)  # CPU使用率
            memory_per = round((psutil.virtual_memory().percent) / 100, 2)  # 内存使用率
            disk_per = round((psutil.disk_usage("/").percent) / 100, 2)  # 硬盘使用率

            read_bytes = psutil.disk_io_counters().read_bytes
            write_bytes = psutil.disk_io_counters().write_bytes
            time.sleep(1)
            read_bytes2 = psutil.disk_io_counters().read_bytes
            write_bytes2 = psutil.disk_io_counters().write_bytes
            Read_bytes = read_bytes2 - read_bytes  # 硬盘读的速率bps
            Write_bytes = write_bytes2 - write_bytes  # 硬盘写的速率bps

            hostname = socket.gethostname()  # 获取本机计算机名称
            ip = socket.gethostbyname(hostname)  # 获取本机ip

            port = ENGINE_RUN_PORT
            data = {'cpu': cpu_per, 'memory': memory_per, 'disk': disk_per, 'readBps': Read_bytes, 'writeBps': Write_bytes,
                    'ip': ip, 'port': port, 'recordTime': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))}
            # print(data)
            response = requests.post(PERFORMANCE_UPDATE_URL, headers=HEADERS, timeout=10, data=json.dumps(data))
            # print('performance', response)

        except Exception as e:
            # print(e)
            # print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())), e)
            pass
        if time.time() - time_start < WAIT_TIME_MONITOR:
            time.sleep(WAIT_TIME_MONITOR - int(time.time() - time_start))


def GPU_monitoring_task():
    while True:  # 轮询执行任务
        time_start = time.time()
        try:
            GPUs = GPUtil.getGPUs()  # 获取所有可用GPU列表，id号从0开始
            # GPUavailability = GPUtil.getAvailability(GPUs)  #返回一个与GPUs大小相等的1和0列表，指示哪些对应的GPU可用，1表示可用，0表示不可用
            gpu_per_list = []
            for i, gpu in enumerate(GPUs):
                gpu_util = gpu.memoryUtil
                gpu_per_list.append(gpu_util)
            if len(gpu_per_list) == 0:
                gpu_per = 0
            else:
                gpu_per = sum(gpu_per_list) / len(gpu_per_list)
            gpu_per = round(gpu_per, 2)
            hostname = socket.gethostname()  # 获取本机计算机名称
            ip = socket.gethostbyname(hostname)  # 获取本机ip

            port = ENGINE_RUN_PORT
            data = {'useRatio': gpu_per, 'ip': ip, 'port': port, 'recordTime': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))}
            # print(data)
            response = requests.post(GPU_UPDATE_URL, headers=HEADERS, timeout=10, data=json.dumps(data))
            # print('GPU', response)

        except Exception as e:
            # print(e)
            # print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())), e)
            pass
        if time.time() - time_start < WAIT_TIME_MONITOR:
            time.sleep(WAIT_TIME_MONITOR - int(time.time() - time_start))


class aicv_Logger(object):
    def __init__(self, fileN="Default.log"):
        # self.terminal = sys.stdout
        self.log = open(fileN, "a")

    def write(self, message):
        # self.terminal.write(message)
        self.log.write(message)

    def flush(self):
        pass


def save_log(content):
    f = open(os.path.join(os.path.dirname(os.path.realpath(__file__)),"log/aicv.txt"), "a")
    print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + str(content), file=f, flush=True)


if __name__ == "__main__":
    preprocess_task = mp.Process(target=execute_preprocess_task)
    labeling_task = mp.Process(target=execute_labeling_task)
    training_task = mp.Process(target=execute_training_task)
    classify_task = mp.Process(target=classify_service_task)
    detect_task = mp.Process(target=detect_service_task)
    segment_task = mp.Process(target=segment_service_task)
    monitoring_task = mp.Process(target=performance_monitoring_task)
    GPU_monitor_task = mp.Process(target=GPU_monitoring_task)

    preprocess_task.start()
    labeling_task.start()
    training_task.start()
    classify_task.start()
    detect_task.start()
    segment_task.start()
    monitoring_task.start()
    GPU_monitor_task.start()

    hostname = socket.gethostname()  # 获取本机计算机名称
    ip = socket.gethostbyname(hostname)  # 获取本机ip
    print(ip)
    save_log(' IP:' + ip)
    uvicorn.run("engine:app", host=str(ip), port=ENGINE_RUN_PORT)
